package kafka_yl.consumer;

import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class KafkaConsumerTest implements Runnable {

    private final KafkaConsumer<String, String> consumer;
    private ConsumerRecords<String, String> msgList;
    private final String topic;
    private static final String GROUPID = "groupA";

    public KafkaConsumerTest(String topicName) {
        Properties props = new Properties();
        //bootstrap.servers： kafka的地址
        props.put("bootstrap.servers", "192.168.19.161:6667");
        //group.id：组名 不同组名可以重复消费
        props.put("group.id", GROUPID);
        //enable.auto.commit：是否自动提交，默认为true
  		props.put("enable.auto.commit", "true");
  		//auto.commit.interval.ms: 从poll(拉)的回话处理时长
        props.put("auto.commit.interval.ms", "1000");
        //session.timeout.ms:超时时间
  		props.put("session.timeout.ms", "30000");
  		//max.poll.records:一次最大拉取的条数
  		props.put("max.poll.records", 1000);
  		//auto.offset.reset：消费规则，默认earliest 
  		//earliest，当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
  		//latest，当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 
  		//none，topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
        props.put("auto.offset.reset", "earliest");
  		//key.serializer: 键序列化
        props.put("key.deserializer", StringDeserializer.class.getName());
  		//value.deserializer:值序列化
	    props.put("value.deserializer", StringDeserializer.class.getName());
	    this.consumer = new KafkaConsumer<String, String>(props);
	    this.topic = topicName;
  		//订阅一个topic
        this.consumer.subscribe(Arrays.asList(topic));
    }

    @Override
    public void run() {
    	int messageNo = 1;
    	System.out.println("---------开始消费---------");
		try{
	    	for (;;) {
		    	//抓取100条数据
		    	msgList = consumer.poll(100);
		        if(null!=msgList&&msgList.count()>0){
		        	for (ConsumerRecord<String, String> record : msgList) {
		        		//消费100条就打印 ,但打印的数据不一定是这个规律的
		        		if(messageNo%10==0){
		        			System.out.println(messageNo+"=======receive: key = " + record.key() + ", value = " + record.value()+" offset==="+record.offset());
		        		}
		        		//当消费了1000条就退出
		        		if(messageNo%100==0){
		        			break;
		        		}
		        		messageNo++;
		        	}
		        }
		        else{  
		        	Thread.sleep(1000);
	          	}
	    	}       
	      } catch (InterruptedException e) {
	          e.printStackTrace();
	      } finally {
	          consumer.close();
	      }
    } 
    
 	public static void main(String args[]) {
 		KafkaConsumerTest test = new KafkaConsumerTest("KAFKA_YL");
 		Thread thread = new Thread(test);
 		thread.start();
 	}
}
